package drds.binlog.parse.inbound.mysql;

import com.lmax.disruptor.EventHandler;
import com.lmax.disruptor.LifecycleAware;
import drds.binlog.LogEvent;
import drds.binlog.parse.exception.BinlogParseException;


class SinkStoreStage implements EventHandler<MessageEvent>, LifecycleAware
{

    private MultiStageCoprocessor mysqlMultiStageCoprocessor;

    public SinkStoreStage(MultiStageCoprocessor mysqlMultiStageCoprocessor)
    {
        this.mysqlMultiStageCoprocessor = mysqlMultiStageCoprocessor;
    }

    public void onEvent(MessageEvent messageEvent, long sequence, boolean endOfBatch) throws Exception
    {
        try
        {
            if (messageEvent.getEntry() != null)
            {
                mysqlMultiStageCoprocessor.eventTransactionBuffer.add(messageEvent.getEntry());
            }

            LogEvent logEvent = messageEvent.getLogEvent();
            if (mysqlMultiStageCoprocessor.connection instanceof MysqlConnection && logEvent.getSemival() == 1)
            {
                // semi ack回报
                ((MysqlConnection) mysqlMultiStageCoprocessor.connection).sendSemiAck(logEvent.getLogHeader().getLogFileName(),
                        logEvent.getLogHeader().getLogPosition());
            }

            // clear for gc
            messageEvent.setLogByteBlock(null);
            messageEvent.setLogEvent(null);
            messageEvent.setTableMetaData(null);
            messageEvent.setEntry(null);
            messageEvent.setNeedDmlParse(false);
        } catch (Throwable e)
        {
            mysqlMultiStageCoprocessor.binlogParseException = new BinlogParseException(e);
            throw mysqlMultiStageCoprocessor.binlogParseException;
        }
    }

    @Override
    public void onStart()
    {

    }

    @Override
    public void onShutdown()
    {

    }
}
